Amazon S3 Tables を AWS Glue 用いてNamespaceとテーブルの作成、データ追加・削除までやってみた #AWSreInvent

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from pyspark.conf import SparkConf  

warehouse = "arn:aws:s3tables:us-east-1:123456789012:bucket/cm-namespace-20241218"

conf = SparkConf()
conf.set("spark.sql.catalog.s3tablesbucket", "org.apache.iceberg.spark.SparkCatalog")
conf.set("spark.sql.catalog.s3tablesbucket.catalog-impl", "software.amazon.s3tables.iceberg.S3TablesCatalog")
conf.set("spark.sql.catalog.s3tablesbucket.warehouse", warehouse)
conf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")

sparkContext = SparkContext(conf=conf)
glueContext = GlueContext(sparkContext)

# Spark Session
spark = glueContext.spark_session

# Create namespace in S3 Tables
spark.sql( """ CREATE NAMESPACE IF NOT EXISTS s3tablesbucket.cm_namespace """)           
# # Drop Table in S3 Tables
# spark.sql( """ DROP TABLE IF EXISTS s3tablesbucket.cm_namespace.`users` """)
# # But "Error Category: UNSUPPORTED_OPERATION_ERROR; Failed Line Number: 26; UnsupportedOperationException: S3 Tables does not support the dropTable operation with purge=false. Some versions of Spark always set this flag to false even when running DROP TABLE PURGE commands. You can retry with DROP TABLE PURGE or use the S3 Tables DeleteTable API to delete a table."

# Delete in S3 Tables
spark.sql( """ DELETE FROM s3tablesbucket.cm_namespace.`users` """)

# Create Table using Iceberg in S3 Tables
spark.sql( """ CREATE TABLE IF NOT EXISTS s3tablesbucket.cm_namespace.`users` (id INT, name STRING) USING iceberg """)

# Use Namespace
spark.sql( """ USE s3tablesbucket.cm_namespace """).show()

# Show Tables
spark.sql( """ SHOW TABLES """).show()

# Insert records into Iceberg table in S3 Tables
spark.sql( """ INSERT INTO s3tablesbucket.cm_namespace.`users` (id, name) VALUES(1, 'Alice')""" )

# Read from Iceberg table in S3 Tables
spark.sql(""" SELECT * FROM s3tablesbucket.cm_namespace.`users` """).show()